package org.example.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.example.config.RocketMQConfigProperties;
import org.example.utils.TmpTools;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Set;

/**
 * @author c3886
 */
@Slf4j
@RestController
@RequestMapping("/consumer")
public class PullController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private RocketMQConfigProperties prop;

    @RequestMapping("/sendDataToPull")
    public String sendDataToPull() {
        rocketMQTemplate.syncSend(prop.getTopic().get("pull"), "sendDataToPull");
        return "ok";
    }

    @RequestMapping("/pull")
    public String pull() throws Exception {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer(prop.getConsumer().getGroup().get("pull"));
        consumer.setNamesrvAddr(prop.getNameServer());
        consumer.start();
        Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues(prop.getTopic().get("pull"));
        for(MessageQueue queue : messageQueues){
            // 从偏移量0开始拉取所有的数据
            long offset = 0;
            while(true){
                PullResult pullResult = consumer.pull(queue, "*", offset, 1);
                offset = pullResult.getNextBeginOffset();
                // 在队列中拉取不到消息就结束
                if(PullStatus.FOUND != pullResult.getPullStatus()) {
                    break;
                }
                List<MessageExt> messageExtList = pullResult.getMsgFoundList();
                for (MessageExt m : messageExtList) {
                    log.info("当前队列为："+queue.getQueueId()+"，偏移量为："+offset+
                            "，拉取到数据："+ new String(m.getBody())+"，投递的时间为：" + TmpTools.timeToString(m.getBornTimestamp()));
                }
            }
        }
        consumer.shutdown();
        return "ok";
    }

}
